//package com.sunye.kafka;
//
//import com.alibaba.fastjson.JSONObject;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.annotation.PartitionOffset;
//import org.springframework.kafka.annotation.TopicPartition;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.kafka.support.KafkaHeaders;
//import org.springframework.messaging.handler.annotation.Header;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//import java.util.Map;
//
//@Component
//public class KafkaListenController {
//
//    //监听topic 消费者组是linux中设置
//    @KafkaListener(topics = "topic1",groupId = "test-consumer-group")
//    public void listenMessage(ConsumerRecord<?,?> record, Acknowledgment ack,
//                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
//
//        //topic 监听的主题    partition分区   offset偏移量
//        System.out.println("消费消息："+record.topic()+"---"+record.partition()+"---"+record.offset()+"---"+record.value());
//        Map map = JSONObject.parseObject(record.value().toString(), Map.class);
//        List<String> list =(List<String>) map.get("list");
//        System.out.println("list = " + list);
//        //手动确认消费成功  消息才从队列中消失
//        ack.acknowledge();
//        //拒绝消费
//        //ack.nack(1);
//    }
//
//    //监听 配置分区
//    @KafkaListener(topicPartitions =
//            { @TopicPartition(topic = "topicWithCall",
//                    partitions = { "0", "2","3" },//监听0、2、3分区
//                    partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "4"))//监听1分区里面offset从4开始的消息
//            },groupId = "test-consumer-group"
//    )
//    public void topicWithCall(ConsumerRecord<?,?> record, Acknowledgment ack,
//                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
//
//        //topic 监听的路由    partition分区   offset偏移量
//        System.out.println("消费消息："+record.topic()+"---"+record.partition()+"---"+record.offset()+"---"+record.value());
//        //手动确认消费成功
//        ack.acknowledge();
//    }
//
//
////    //监听自定义分区
////    @KafkaListener(topics = "customizeTopic",groupId = "test-consumer-group")
////    public void customizePartition(ConsumerRecord<?,?> record, Acknowledgment ack){
////        //topic 监听的路由    partition分区   offset偏移量
////        System.out.println("消费消息："+record.topic()+"---"+record.partition()+"---"+record.offset()+"---"+record.value());
////        //手动确认消费成功
////        ack.acknowledge();
////    }
//
//    //监听topicTrans 消费者组是linux中设置
//    @KafkaListener(topics = "topicTrans",groupId = "test-consumer-group")
//    public void listenTransMessage(ConsumerRecord<?,?> record, Acknowledgment ack,
//                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
//        System.out.println("消费消息："+record.topic()+"---"+record.partition()+"---"+record.value());
//        //手动确认消费成功
//        ack.acknowledge();
//    }
//
//    //通配 .*
//    @KafkaListener(topicPattern = "topic.*",groupId = "test-consumer-group")
//    public void listenPatternMessage(ConsumerRecord<?,?> record, Acknowledgment ack,
//                                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
//        System.out.println("通配消费："+record.topic()+"---"+record.partition()+"---"+record.value());
//        //手动确认消费成功
//        ack.acknowledge();
//    }
//}
